package com.wafersystems.cloud.service.impl.devicecontroller.modbus;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import com.alibaba.fastjson.JSON;
import com.wafersystems.cloud.bo.abstractbo.AbstractSwitchDevice;
import com.wafersystems.cloud.service.impl.DeviceService;
import com.wafersystems.cloud.vo.MqMsg;

import lombok.extern.slf4j.Slf4j;

/**
 * 康奈德 modbus服务端
 * 
 * @author lihy
 */
@Slf4j
public class ServerHandle implements Runnable {
  private Selector selector;
  private ServerSocketChannel serverChannel;
  private volatile boolean started;
  public static ArrayBlockingQueue<byte[]> writeMsg = new ArrayBlockingQueue<byte[]>(10, true);
  private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

  private DeviceService deviceService;

  private final int port = 3388;

  public void setDeviceService(DeviceService deviceService) {
    this.deviceService = deviceService;
  }

  /**
   * 构造方法
   * 
   * @param port 指定要监听的端口号
   */
  public ServerHandle() {
    try {
      // 创建选择器
      selector = Selector.open();
      // 打开监听通道
      serverChannel = ServerSocketChannel.open();
      // 如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式
      serverChannel.configureBlocking(false);// 开启非阻塞模式
      // 绑定端口 backlog设为1024
      serverChannel.socket().bind(new InetSocketAddress(port), 1024);
      // 监听客户端连接请求
      serverChannel.register(selector, SelectionKey.OP_ACCEPT);
      // 标记服务器已开启
      started = true;
      log.info("服务器已启动，端口号：" + port);
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  public void stop() {
    started = false;
  }

  @Override
  public void run() {
    // 循环遍历selector
    while (started) {
      try {
        // 无论是否有读写事件发生，selector每隔1s被唤醒一次
        selector.select(1);
        // 阻塞,只有当至少一个注册的事件发生的时候才会继续.
        // selector.select();
        Set<SelectionKey> keys = selector.selectedKeys();
        Iterator<SelectionKey> it = keys.iterator();
        SelectionKey key = null;
        while (it.hasNext()) {
          key = it.next();
          it.remove();
          try {
            handleInput(key);
          } catch (Exception e) {
            if (key != null) {
              key.cancel();
              if (key.channel() != null) {
                key.channel().close();
              }
            }
          }
        }
      } catch (Throwable t) {
        t.printStackTrace();
      }
    }
    // selector关闭后会自动释放里面管理的资源
    if (selector != null)
      try {
        selector.close();
      } catch (Exception e) {
        e.printStackTrace();
      }
  }

  private void handleInput(SelectionKey key) throws IOException {
    if (key.isValid()) {
      // 处理新接入的请求消息
      if (key.isAcceptable()) {
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        // 通过ServerSocketChannel的accept创建SocketChannel实例
        // 完成该操作意味着完成TCP三次握手，TCP物理链路正式建立
        SocketChannel sc = ssc.accept();
        // 设置为非阻塞的
        sc.configureBlocking(false);
        // 注册为读
        sc.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
      }

      if (key.isReadable()) {
        SocketChannel sc = (SocketChannel) key.channel();
        // 创建ByteBuffer，并开辟一个1M的缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(24);
        // 读取请求码流，返回读取到的字节数
        int readBytes = sc.read(buffer);
        // 读取到字节，对字节进行编解码
        if (readBytes > 0) {
          String clientIp = sc.socket().getLocalAddress().toString();
          log.info("客户端ip：" + clientIp);
          // 将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作
          buffer.flip();
          // 根据缓冲区可读字节数创建字节数组
          byte[] bytes = new byte[buffer.remaining()];
          // 将缓冲区可读字节数组复制到新建的数组中
          buffer.get(bytes);

          String str = "";
          for (int i = 0; i < bytes.length; i++)
            str += bytes[i] + ",";
          log.info(str);
          // 处理数据
          try {
            log.info("事务处理标识：" + (bytes[0] + bytes[1]));
            log.info("协议标识：" + (bytes[2] + bytes[3]));
            log.info("后面字节数：" + (bytes[4] + bytes[5]));
            log.info("单元标识：" + bytes[6]);
            log.info("功能码：" + bytes[7]);
            log.info("数据长度：" + bytes[8]);
            int dataLength = bytes[8];
            String lineNumBinaryStrAllPort = "";
            int index = 8 + dataLength;
            for (int i = 0; i < dataLength; i++) {
              lineNumBinaryStrAllPort += toBinaryString(bytes[index--]);
            }
            log.info("收到的原始数据 {}", lineNumBinaryStrAllPort);
            int duanziIndex = 1;
            for (int i = lineNumBinaryStrAllPort.length(); i > 0; i = i - 1) {
              if ("0".equals(lineNumBinaryStrAllPort.substring(i - 1, i))) {
                log.info("端子：{}  断电---", duanziIndex);
                notify(duanziIndex, (int) bytes[6], clientIp, 0);
              } else {
                log.info("端子：{}  加电+++", duanziIndex);
                notify(duanziIndex, (int) bytes[6], clientIp, 1);
              }
              duanziIndex++;
            }
            /*
             * List<String> lineNumBinaryStrs = getAllLineNumBinaryStr(lineNumBinaryStrAllPort); for
             * (String lineNumBinaryStr : lineNumBinaryStrs) { log.info("---" + lineNumBinaryStr);
             * for (int i = 0; i <= dataLength * 8; i++) { if
             * (lineNumBinaryStr.endsWith(Integer.toBinaryString((1 << i)))) { log.info("端子：{} 加电。",
             * (i + 1)); notify(i + 1, (int) bytes[6], clientIp); } } }
             */
          } catch (Exception e) {
            log.error(e.getMessage(), e);
          }
        }
        // 没有读取到字节 忽略
        // else if(readBytes==0);
        // 链路已经关闭，释放资源
        else if (readBytes < 0) {
          key.cancel();
          sc.close();
        }

      } else if (key.isWritable()) {
        log.debug("Writeable event happened");
        if (writeMsg.size() > 0) {
          send(key);
        }
      }
    }
  }

  private void send(SelectionKey selectionKey) {
    Runnable task = new Runnable() {
      public void run() {
        try {
          log.info("send command");
          byte[] searchByte = writeMsg.take();

          StringBuffer bytelog = new StringBuffer();
          for (byte b : searchByte) {
            bytelog.append((b & 0xff)).append(" ");
          }

          if (selectionKey.isValid()) {
            if (selectionKey.isWritable()) {
              log.info("send byte[] {}", bytelog);
              SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
              // 3、指定缓冲区大小
              ByteBuffer byteBuffer = ByteBuffer.allocate(searchByte.length);
              byteBuffer.put(searchByte);
              // 4、切换到读取模式
              byteBuffer.clear();
              // 5、写入到缓冲区
              socketChannel.write(byteBuffer);
              byteBuffer.compact();
            }
          }
        } catch (Exception e) {
          log.error(e.getMessage(), e);
        }
      }
    };
    executor.execute(task);
  }

  /**
   * 
   * @param port 康奈德设备接线口
   * @param deviceId 康奈德设备唯一标识 （1-127）
   * @param clientIp 康奈德设备ip
   * @param i
   */
  private void notify(int socketId, int deviceId, String clientIp, int switchValue) {
    AbstractSwitchDevice device = deviceService.getDiDevice(socketId + "");
    if (null != device) {
      MqMsg msg = new MqMsg();
      msg.setSocketId(socketId + "");
      msg.setValue(switchValue + "");
      msg.setHostId(deviceId + "");
      msg.setSupplier("康奈德");
      device.sendMsg(JSON.toJSONString(msg));
    }
  }

  /*
   * private List<String> getAllLineNumBinaryStr(String lineNumBinaryStr) { List<String> allPorts =
   * new ArrayList<String>(); for (int i = 0; i < lineNumBinaryStr.length(); i++) { char port =
   * lineNumBinaryStr.charAt(i); if (port == '1') { StringBuffer onePort = new StringBuffer(); for
   * (int j = 0; j < lineNumBinaryStr.length(); j++) { if (j == i) { onePort.append(1); } else {
   * onePort.append(0); } } allPorts.add(onePort.toString()); } } return allPorts; }
   */

  /**
   * 转为2进制 不足8位补0
   * 
   * @param b
   * @return
   */
  public static String toBinaryString(byte b) {
    String binary = Integer.toBinaryString(b);
    binary = String.format("%08d", Integer.parseInt(binary));
    return binary;
  }
}
